package com.demo.branch;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.TreeMap;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 大文件统计频数
 * 采用分治算法，将大文件中数字按照文件个数和指定范围拆分为小文件
 * 统计小文件中数字的出现次数，再统计所有文件结果中出现次数最多的结果
 * 参考：https://blog.csdn.net/qq_26437925/article/details/78531179
 */
public class TopK {

    static String fileLoc = "D:/topk/nums.txt";
    static int TOP_K = 10;
    static int FILE_CNT = 500;

    public static void main(String[] args) {
        long startTime = System.currentTimeMillis();
        generateBigFile();
        System.out.println("创建文件：" + (System.currentTimeMillis() - startTime));

        startTime = System.currentTimeMillis();
        hashToSmallFiles();
        System.out.println("拆分文件：" + (System.currentTimeMillis() - startTime));

        startTime = System.currentTimeMillis();
        Map<Integer, List<Content>> map = countEverySmallFile();
        List<Content> list = calculateResult(map);
        list.forEach(content -> System.out.println(content.getContent() + "\t" + content.getCount()));
        System.out.println("统计文件：" + (System.currentTimeMillis() - startTime));
    }

    // 生成原始大文件
    public static void generateBigFile() {
        try {
            File file = new File(fileLoc);
            if (!file.exists()) {
                boolean newFile = file.createNewFile();
                System.out.println("创建文件完成");
            }
            if (file.length() < 1) {
                FileWriter fileWriter = new FileWriter(file);
                BufferedWriter bufferedWriter = new BufferedWriter(fileWriter);
                Random random = new Random();
                for (int i = 0; i < 1024 * 1024 * 1024; i++) {
                    // 产生【0,10000】之间的随机数
                    int randint = (int) Math.floor(random.nextDouble() * 100000.0);
                    bufferedWriter.write(String.valueOf(randint));
                    bufferedWriter.newLine();
                }
                bufferedWriter.close();
                fileWriter.close();
                System.out.println("文件写入完成");
            }
            System.out.println((file.length() / (1024.0 * 1024.0 * 1024.0)) + " G");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    // 拆分成小文件
    private static void hashToSmallFiles() {
        try {
            BufferedReader bufferedReader = new BufferedReader(new FileReader(fileLoc));
            HashMap<String, FileWriter> fileWriters = new HashMap<>();
            String line;
            while ((line = bufferedReader.readLine()) != null) {
                int tmp = Math.abs(line.hashCode() % FILE_CNT);
                String fileName = fileLoc.replace(".txt", tmp + ".txt");
                FileWriter fileWriter;
                if (fileWriters.containsKey(fileName)) {
                    fileWriter = fileWriters.get(fileName);
                } else {
                    fileWriter = new FileWriter(fileName, true);
                    fileWriters.put(fileName, fileWriter);
                }
                fileWriter.write(line + "\n");
            }
            bufferedReader.close();
            for (FileWriter item : fileWriters.values()) {
                item.close();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    static Map<Integer, List<Content>> contentMap = new ConcurrentHashMap<>();

    static class sumThread implements Runnable {

        int index;

        public sumThread(int index) {
            this.index = index;
        }

        // 多线程，统计单个小文件频数
        @Override
        public void run() {
            try {
                List<Content> list = new ArrayList<>();
                File file = new File(fileLoc.replace(".txt", index + ".txt"));
                if (!file.exists()) return;
                long startTime = System.currentTimeMillis();
                BufferedReader bufferedReader = new BufferedReader(new FileReader(file));
                String line;
                // 1.统计频次
                HashMap<String, Integer> sumMap = new HashMap<>();
                while ((line = bufferedReader.readLine()) != null) {
                    if (!sumMap.containsKey(line)) {
                        sumMap.put(line, 1);
                    } else {
                        sumMap.put(line, sumMap.get(line) + 1);
                    }
                }
                // 2.结果排序
                TreeMap<Integer, String> treeMap = new TreeMap<>((a, b) -> -(a - b));
                for (String content : sumMap.keySet()) {
                    int num = sumMap.get(content);
                    treeMap.put(num, content);
                }
                // 获取前TOP_K个数据，如果第TOP_K个后续有频次重复的数据也需要加入
                int total = 0;
                for (Integer cnt : treeMap.keySet()) {
                    Content content = new Content(treeMap.get(cnt), cnt);
                    total++;
                    if (total > TOP_K) {
                        if (total == list.get(TOP_K - 1).getCount()) {
                            list.add(content);
                        } else {
                            break;
                        }
                    } else {
                        list.add(content);
                    }
                }
                contentMap.put(index, list);
                System.out.println("统计文件" + fileLoc.replaceAll(".txt", index + ".txt") + "使用" + (System.currentTimeMillis() - startTime) + "毫秒");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    private static Map<Integer, List<Content>> countEverySmallFile() {
        try {
            Thread[] threads = new Thread[FILE_CNT];
            for (int i = 0; i < FILE_CNT; i++) {
                threads[i] = new Thread(new sumThread(i));
            }
            for (int i = 0; i < FILE_CNT; i++) {
                threads[i].start();
            }
            for (int i = 0; i < FILE_CNT; i++) {
                threads[i].join();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return contentMap;
    }

    private static List<Content> calculateResult(Map<Integer, List<Content>> sumList) {
        TreeMap<Integer, String> temp = new TreeMap<>((a, b) -> -(a - b));
        for (List<Content> list : sumList.values()) {
            for (Content content : list) {
                temp.put(content.getCount(), content.getContent());
            }
        }
        int index = 0;
        List<Content> list = new ArrayList<>();
        for (Map.Entry<Integer, String> item : temp.entrySet()) {
            if (index > TOP_K) break;
            list.add(new Content(item.getValue(), item.getKey()));
            index++;
        }
        return list;
    }

}

class Content implements Serializable {
    private String content;
    private int count;

    public Content(String content, int count) {
        this.content = content;
        this.count = count;
    }

    public String getContent() {
        return content;
    }

    public void setContent(String content) {
        this.content = content;
    }

    public int getCount() {
        return count;
    }

    public void setCount(int count) {
        this.count = count;
    }
}
